/*
 * Copyright 2012 The Netty Project
 *
 * The Netty Project licenses this file to you under the Apache License,
 * version 2.0 (the "License"); you may not use this file except in compliance
 * with the License. You may obtain a copy of the License at:
 *
 *   http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations
 * under the License.
 */
package io.netty.util.concurrent;

import io.netty.util.internal.ObjectUtil;
import io.netty.util.internal.PlatformDependent;
import io.netty.util.internal.SystemPropertyUtil;
import io.netty.util.internal.UnstableApi;
import io.netty.util.internal.logging.InternalLogger;
import io.netty.util.internal.logging.InternalLoggerFactory;

import java.lang.Thread.State;
import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;

/**
 * Abstract base class for {@link OrderedEventExecutor}'s that execute all its submitted tasks in a
 * single thread.
 */
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements
    OrderedEventExecutor {

  static final int DEFAULT_MAX_PENDING_EXECUTOR_TASKS = Math.max(16,
      SystemPropertyUtil.getInt("io.netty.eventexecutor.maxPendingTasks", Integer.MAX_VALUE));

  private static final InternalLogger logger =
      InternalLoggerFactory.getInstance(SingleThreadEventExecutor.class);

  private static final int ST_NOT_STARTED = 1;
  private static final int ST_STARTED = 2;
  private static final int ST_SHUTTING_DOWN = 3;
  private static final int ST_SHUTDOWN = 4;
  private static final int ST_TERMINATED = 5;

  private static final Runnable WAKEUP_TASK = new Runnable() {
    @Override
    public void run() {
      // Do nothing.
    }
  };
  private static final Runnable NOOP_TASK = new Runnable() {
    @Override
    public void run() {
      // Do nothing.
    }
  };

  private static final AtomicIntegerFieldUpdater<SingleThreadEventExecutor> STATE_UPDATER =
      AtomicIntegerFieldUpdater.newUpdater(SingleThreadEventExecutor.class, "state");
  private static final AtomicReferenceFieldUpdater<SingleThreadEventExecutor, ThreadProperties> PROPERTIES_UPDATER =
      AtomicReferenceFieldUpdater.newUpdater(
          SingleThreadEventExecutor.class, ThreadProperties.class, "threadProperties");

  // 这里的taskQueue默认是LinkedBlockingQueue
  private final Queue<Runnable> taskQueue;

  private volatile Thread thread;
  @SuppressWarnings("unused")
  private volatile ThreadProperties threadProperties;
  private final Executor executor;
  private volatile boolean interrupted;

  private final Semaphore threadLock = new Semaphore(0);
  private final Set<Runnable> shutdownHooks = new LinkedHashSet<Runnable>();
  private final boolean addTaskWakesUp;
  private final int maxPendingTasks;
  private final RejectedExecutionHandler rejectedExecutionHandler;

  private long lastExecutionTime;

  @SuppressWarnings({"FieldMayBeFinal", "unused"})
  private volatile int state = ST_NOT_STARTED;

  private volatile long gracefulShutdownQuietPeriod;
  private volatile long gracefulShutdownTimeout;
  private long gracefulShutdownStartTime;

  private final Promise<?> terminationFuture = new DefaultPromise<Void>(
      GlobalEventExecutor.INSTANCE);

  /**
   * Create a new instance
   *
   * @param parent the {@link EventExecutorGroup} which is the parent of this instance and belongs
   * to it
   * @param threadFactory the {@link ThreadFactory} which will be used for the used {@link Thread}
   * @param addTaskWakesUp {@code true} if and only if invocation of {@link #addTask(Runnable)} will
   * wake up the executor thread
   */
  protected SingleThreadEventExecutor(
      EventExecutorGroup parent, ThreadFactory threadFactory, boolean addTaskWakesUp) {
    this(parent, new ThreadPerTaskExecutor(threadFactory), addTaskWakesUp);
  }

  /**
   * Create a new instance
   *
   * @param parent the {@link EventExecutorGroup} which is the parent of this instance and belongs
   * to it
   * @param threadFactory the {@link ThreadFactory} which will be used for the used {@link Thread}
   * @param addTaskWakesUp {@code true} if and only if invocation of {@link #addTask(Runnable)} will
   * wake up the executor thread
   * @param maxPendingTasks the maximum number of pending tasks before new tasks will be rejected.
   * @param rejectedHandler the {@link RejectedExecutionHandler} to use.
   */
  protected SingleThreadEventExecutor(
      EventExecutorGroup parent, ThreadFactory threadFactory,
      boolean addTaskWakesUp, int maxPendingTasks, RejectedExecutionHandler rejectedHandler) {
    this(parent, new ThreadPerTaskExecutor(threadFactory), addTaskWakesUp, maxPendingTasks,
        rejectedHandler);
  }

  /**
   * Create a new instance
   *
   * @param parent the {@link EventExecutorGroup} which is the parent of this instance and belongs
   * to it
   * @param executor the {@link Executor} which will be used for executing
   * @param addTaskWakesUp {@code true} if and only if invocation of {@link #addTask(Runnable)} will
   * wake up the executor thread
   */
  protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
      boolean addTaskWakesUp) {
    this(parent, executor, addTaskWakesUp, DEFAULT_MAX_PENDING_EXECUTOR_TASKS,
        RejectedExecutionHandlers.reject());
  }

  /**
   * Create a new instance
   *
   * @param parent the {@link EventExecutorGroup} which is the parent of this instance and belongs
   * to it
   * @param executor the {@link Executor} which will be used for executing
   * @param addTaskWakesUp {@code true} if and only if invocation of {@link #addTask(Runnable)} will
   * wake up the executor thread
   * @param maxPendingTasks the maximum number of pending tasks before new tasks will be rejected.
   * @param rejectedHandler the {@link RejectedExecutionHandler} to use.
   */
  protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor,
      boolean addTaskWakesUp, int maxPendingTasks,
      RejectedExecutionHandler rejectedHandler) {
    super(parent);
    this.addTaskWakesUp = addTaskWakesUp;
    this.maxPendingTasks = Math.max(16, maxPendingTasks);
    this.executor = ObjectUtil.checkNotNull(executor, "executor");
    taskQueue = newTaskQueue(this.maxPendingTasks);
    rejectedExecutionHandler = ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler");
  }

  /**
   * @deprecated Please use and override {@link #newTaskQueue(int)}.
   */
  @Deprecated
  protected Queue<Runnable> newTaskQueue() {
    return newTaskQueue(maxPendingTasks);
  }

  /**
   * Create a new {@link Queue} which will holds the tasks to execute. This default implementation
   * will return a {@link LinkedBlockingQueue} but if your sub-class of {@link
   * SingleThreadEventExecutor} will not do any blocking calls on the this {@link Queue} it may make
   * sense to {@code @Override} this and return some more performant implementation that does not
   * support blocking operations at all.
   */
  protected Queue<Runnable> newTaskQueue(int maxPendingTasks) {
    return new LinkedBlockingQueue<Runnable>(maxPendingTasks);
  }

  /**
   * Interrupt the current running {@link Thread}.
   */
  protected void interruptThread() {
    Thread currentThread = thread;
    if (currentThread == null) {
      interrupted = true;
    } else {
      currentThread.interrupt();
    }
  }

  /**
   * @see Queue#poll()
   */
  protected Runnable pollTask() {
    assert inEventLoop();
    return pollTaskFrom(taskQueue);
  }

  protected static Runnable pollTaskFrom(Queue<Runnable> taskQueue) {
    for (; ; ) {
      // 这里poll方法会取出队列头部的一个任务，并且是非阻塞的，不存在则返回null
      Runnable task = taskQueue.poll();
      if (task == WAKEUP_TASK) {
        continue;
      }
      return task;
    }
  }

  /**
   * Take the next {@link Runnable} from the task queue and so will block if no task is currently
   * present.
   * <p>
   * Be aware that this method will throw an {@link UnsupportedOperationException} if the task
   * queue, which was created via {@link #newTaskQueue()}, does not implement {@link
   * BlockingQueue}.
   * </p>
   *
   * @return {@code null} if the executor thread has been interrupted or waken up.
   */
  protected Runnable takeTask() {
    assert inEventLoop();
    if (!(taskQueue instanceof BlockingQueue)) {
      throw new UnsupportedOperationException();
    }

    // 这里的主要作用就是通过轮询的方式不断的从scheduledTaskQueue取出task，如果task不为空，则将其放到taskQueue中，
    // 然后不断的从taskQueue中取出任务返回
    BlockingQueue<Runnable> taskQueue = (BlockingQueue<Runnable>) this.taskQueue;
    for (; ; ) {
      ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
      if (scheduledTask == null) {
        Runnable task = null;
        try {
          task = taskQueue.take();
          if (task == WAKEUP_TASK) {
            task = null;
          }
        } catch (InterruptedException e) {
          // Ignore
        }
        return task;
      } else {
        long delayNanos = scheduledTask.delayNanos();
        Runnable task = null;
        if (delayNanos > 0) {
          try {
            task = taskQueue.poll(delayNanos, TimeUnit.NANOSECONDS);
          } catch (InterruptedException e) {
            // Waken up.
            return null;
          }
        }
        if (task == null) {
          // We need to fetch the scheduled tasks now as otherwise there may be a chance that
          // scheduled tasks are never executed if there is always one task in the taskQueue.
          // This is for example true for the read task of OIO Transport
          // See https://github.com/netty/netty/issues/1614
          fetchFromScheduledTaskQueue();
          task = taskQueue.poll();
        }

        if (task != null) {
          return task;
        }
      }
    }
  }

  private boolean fetchFromScheduledTaskQueue() {
    // 这里的nanoTime得到的是当前类被加载的时间到当前时间的一个时间间隔
    long nanoTime = AbstractScheduledEventExecutor.nanoTime();
    // 这里pollScheduledTask()方法会从定时任务队列中取出一个任务，然后将其放到taskQueue中，
    // 由于当前线程会不断地从taskQueue中取出任务执行，因而这些任务最终将还是会在线程轮询中执行该任务
    Runnable scheduledTask = pollScheduledTask(nanoTime);
    while (scheduledTask != null) {
      if (!taskQueue.offer(scheduledTask)) {
        // No space left in the task queue add it back to the scheduledTaskQueue so we pick it up again.
        // 如果任务队列满了，则还是将该任务放到定时任务队列中，以便下次取出继续往taskQueue中添加
        scheduledTaskQueue().add((ScheduledFutureTask<?>) scheduledTask);
        return false;
      }
      // 如果添加成功，则继续从定时任务队列中取出下一个任务，将其放到任务队列中
      scheduledTask = pollScheduledTask(nanoTime);
    }
    return true;
  }

  /**
   * @see Queue#peek()
   */
  protected Runnable peekTask() {
    assert inEventLoop();
    return taskQueue.peek();
  }

  /**
   * @see Queue#isEmpty()
   */
  protected boolean hasTasks() {
    assert inEventLoop();
    return !taskQueue.isEmpty();
  }

  /**
   * Return the number of tasks that are pending for processing.
   *
   * <strong>Be aware that this operation may be expensive as it depends on the internal
   * implementation of the
   * SingleThreadEventExecutor. So use it with care!</strong>
   */
  public int pendingTasks() {
    return taskQueue.size();
  }

  /**
   * Add a task to the task queue, or throws a {@link RejectedExecutionException} if this instance
   * was shutdown before.
   */
  protected void addTask(Runnable task) {
    if (task == null) {
      throw new NullPointerException("task");
    }
    if (!offerTask(task)) {
      reject(task);
    }
  }

  final boolean offerTask(Runnable task) {
    if (isShutdown()) {
      reject();
    }
    return taskQueue.offer(task);
  }

  /**
   * @see Queue#remove(Object)
   */
  protected boolean removeTask(Runnable task) {
    if (task == null) {
      throw new NullPointerException("task");
    }
    return taskQueue.remove(task);
  }

  /**
   * Poll all tasks from the task queue and run them via {@link Runnable#run()} method.
   *
   * @return {@code true} if and only if at least one task was run
   */
  protected boolean runAllTasks() {
    assert inEventLoop();
    boolean fetchedAll;
    boolean ranAtLeastOne = false;

    // 执行人呢无队列中的任务，只要执行的至少一个任务，那么就认为是成功执行了，就返回true，否则返回false
    do {
      fetchedAll = fetchFromScheduledTaskQueue();
      if (runAllTasksFrom(taskQueue)) {
        ranAtLeastOne = true;
      }
    } while (!fetchedAll); // keep on processing until we fetched all scheduled tasks.

    if (ranAtLeastOne) {
      lastExecutionTime = ScheduledFutureTask.nanoTime();
    }
    afterRunningAllTasks();
    return ranAtLeastOne;
  }

  /**
   * Runs all tasks from the passed {@code taskQueue}.
   *
   * @param taskQueue To poll and execute all tasks.
   * @return {@code true} if at least one task was executed.
   */
  protected final boolean runAllTasksFrom(Queue<Runnable> taskQueue) {
    Runnable task = pollTaskFrom(taskQueue);
    if (task == null) {
      return false;
    }
    for (; ; ) {
      safeExecute(task);
      task = pollTaskFrom(taskQueue);
      if (task == null) {
        return true;
      }
    }
  }

  /**
   * Poll all tasks from the task queue and run them via {@link Runnable#run()} method.  This method
   * stops running the tasks in the task queue and returns if it ran longer than {@code
   * timeoutNanos}.
   */
  protected boolean runAllTasks(long timeoutNanos) {
    // 这里主要是从定时任务队列中取出当前可以执行的任务，将其放到taskQueue队列中，以便下面运行这些任务
    fetchFromScheduledTaskQueue();
    // 从任务队列中非阻塞式的取出一个任务
    Runnable task = pollTask();
    if (task == null) {
      // 提供的一个hook方法，对于NioEventLoop而言，其继承了SingleThreadEventLoop，而该类则重写了该方法，
      // 主要是将tailQueue中的任务执行完
      afterRunningAllTasks();
      return false;
    }

    // 根据传入的超时时间，计算任务最多可以执行的终止时间
    final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
    long runTasks = 0;
    long lastExecutionTime;
    for (; ; ) {
      safeExecute(task);

      runTasks++;

      // Check timeout every 64 tasks because nanoTime() is relatively expensive.
      // XXX: Hard-coded value - will make it configurable if it is really a problem.
      // 每64个任务执行一次检查，如果当前时间比计算得到的终止时间要大，则说明应该退出循环，
      // 而开始需要执行整体的轮询，从而监控客户端事件。
      // 这里这个判断也只有在任务队列中任务比较多时才会进入，如果任务队列中的任务执行完了，
      // 那么在下面的判断中，其就会跳出循环
      if ((runTasks & 0x3F) == 0) {
        lastExecutionTime = ScheduledFutureTask.nanoTime();
        if (lastExecutionTime >= deadline) {
          break;
        }
      }

      task = pollTask();
      if (task == null) {
        lastExecutionTime = ScheduledFutureTask.nanoTime();
        break;
      }
    }

    afterRunningAllTasks();
    this.lastExecutionTime = lastExecutionTime;
    return true;
  }

  /**
   * Invoked before returning from {@link #runAllTasks()} and {@link #runAllTasks(long)}.
   */
  @UnstableApi
  protected void afterRunningAllTasks() {
  }

  /**
   * Returns the amount of time left until the scheduled task with the closest dead line is
   * executed.
   */
  // 这里delayNanos()方法获取的是定时队列中最近一个任务的执行时间与目标时间的一个比较，如果该任务的deadline
  // 比传入的时间大，则返回deadline与该时间的一个差值，否则返回0。
  // 其实总结来说，就是返回任务队列中第一个任务相对于目标时间的一个延迟时间
  protected long delayNanos(long currentTimeNanos) {
    ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
    if (scheduledTask == null) {
      return SCHEDULE_PURGE_INTERVAL;
    }

    return scheduledTask.delayNanos(currentTimeNanos);
  }

  /**
   * Returns the absolute point in time (relative to {@link #nanoTime()}) at which the the next
   * closest scheduled task should run.
   */
  @UnstableApi
  protected long deadlineNanos() {
    ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
    if (scheduledTask == null) {
      return nanoTime() + SCHEDULE_PURGE_INTERVAL;
    }
    return scheduledTask.deadlineNanos();
  }

  /**
   * Updates the internal timestamp that tells when a submitted task was executed most recently.
   * {@link #runAllTasks()} and {@link #runAllTasks(long)} updates this timestamp automatically, and
   * thus there's usually no need to call this method.  However, if you take the tasks manually
   * using {@link #takeTask()} or {@link #pollTask()}, you have to call this method at the end of
   * task execution loop for accurate quiet period checks.
   */
  protected void updateLastExecutionTime() {
    lastExecutionTime = ScheduledFutureTask.nanoTime();
  }

  /**
   *
   */
  protected abstract void run();

  /**
   * Do nothing, sub-classes may override
   */
  protected void cleanup() {
    // NOOP
  }

  protected void wakeup(boolean inEventLoop) {
    if (!inEventLoop || state == ST_SHUTTING_DOWN) {
      // Use offer as we actually only need this to unblock the thread and if offer fails we do not care as there
      // is already something in the queue.
      taskQueue.offer(WAKEUP_TASK);
    }
  }

  @Override
  public boolean inEventLoop(Thread thread) {
    return thread == this.thread;
  }

  /**
   * Add a {@link Runnable} which will be executed on shutdown of this instance
   */
  public void addShutdownHook(final Runnable task) {
    if (inEventLoop()) {
      shutdownHooks.add(task);
    } else {
      execute(new Runnable() {
        @Override
        public void run() {
          shutdownHooks.add(task);
        }
      });
    }
  }

  /**
   * Remove a previous added {@link Runnable} as a shutdown hook
   */
  public void removeShutdownHook(final Runnable task) {
    if (inEventLoop()) {
      shutdownHooks.remove(task);
    } else {
      execute(new Runnable() {
        @Override
        public void run() {
          shutdownHooks.remove(task);
        }
      });
    }
  }

  private boolean runShutdownHooks() {
    boolean ran = false;
    // Note shutdown hooks can add / remove shutdown hooks.
    while (!shutdownHooks.isEmpty()) {
      List<Runnable> copy = new ArrayList<Runnable>(shutdownHooks);
      shutdownHooks.clear();
      for (Runnable task : copy) {
        try {
          task.run();
        } catch (Throwable t) {
          logger.warn("Shutdown hook raised an exception.", t);
        } finally {
          ran = true;
        }
      }
    }

    if (ran) {
      lastExecutionTime = ScheduledFutureTask.nanoTime();
    }

    return ran;
  }

  @Override
  public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) {
    if (quietPeriod < 0) {
      throw new IllegalArgumentException("quietPeriod: " + quietPeriod + " (expected >= 0)");
    }
    if (timeout < quietPeriod) {
      throw new IllegalArgumentException(
          "timeout: " + timeout + " (expected >= quietPeriod (" + quietPeriod + "))");
    }
    if (unit == null) {
      throw new NullPointerException("unit");
    }

    if (isShuttingDown()) {
      return terminationFuture();
    }

    boolean inEventLoop = inEventLoop();
    boolean wakeup;
    int oldState;
    for (; ; ) {
      if (isShuttingDown()) {
        return terminationFuture();
      }
      int newState;
      wakeup = true;
      oldState = state;
      if (inEventLoop) {
        newState = ST_SHUTTING_DOWN;
      } else {
        switch (oldState) {
          case ST_NOT_STARTED:
          case ST_STARTED:
            newState = ST_SHUTTING_DOWN;
            break;
          default:
            newState = oldState;
            wakeup = false;
        }
      }
      if (STATE_UPDATER.compareAndSet(this, oldState, newState)) {
        break;
      }
    }
    gracefulShutdownQuietPeriod = unit.toNanos(quietPeriod);
    gracefulShutdownTimeout = unit.toNanos(timeout);

    if (ensureThreadStarted(oldState)) {
      return terminationFuture;
    }

    if (wakeup) {
      wakeup(inEventLoop);
    }

    return terminationFuture();
  }

  @Override
  public Future<?> terminationFuture() {
    return terminationFuture;
  }

  @Override
  @Deprecated
  public void shutdown() {
    if (isShutdown()) {
      return;
    }

    boolean inEventLoop = inEventLoop();
    boolean wakeup;
    int oldState;
    for (; ; ) {
      if (isShuttingDown()) {
        return;
      }
      int newState;
      wakeup = true;
      oldState = state;
      if (inEventLoop) {
        newState = ST_SHUTDOWN;
      } else {
        switch (oldState) {
          case ST_NOT_STARTED:
          case ST_STARTED:
          case ST_SHUTTING_DOWN:
            newState = ST_SHUTDOWN;
            break;
          default:
            newState = oldState;
            wakeup = false;
        }
      }
      if (STATE_UPDATER.compareAndSet(this, oldState, newState)) {
        break;
      }
    }

    if (ensureThreadStarted(oldState)) {
      return;
    }

    if (wakeup) {
      wakeup(inEventLoop);
    }
  }

  @Override
  public boolean isShuttingDown() {
    return state >= ST_SHUTTING_DOWN;
  }

  @Override
  public boolean isShutdown() {
    return state >= ST_SHUTDOWN;
  }

  @Override
  public boolean isTerminated() {
    return state == ST_TERMINATED;
  }

  /**
   * Confirm that the shutdown if the instance should be done now!
   */
  protected boolean confirmShutdown() {
    if (!isShuttingDown()) {
      return false;
    }

    if (!inEventLoop()) {
      throw new IllegalStateException("must be invoked from an event loop");
    }

    // 取消定时任务队列中所有任务的执行
    cancelScheduledTasks();

    if (gracefulShutdownStartTime == 0) {
      gracefulShutdownStartTime = ScheduledFutureTask.nanoTime();
    }

    // 执行当前任务队列中的所有任务，只要执行了至少一个任务，那么就认为执行是成功的，
    // 否则就会执行shutdownHook，这里也是只要运行了一个就返回true
    if (runAllTasks() || runShutdownHooks()) {
      if (isShutdown()) {
        // Executor shut down - no new tasks anymore.
        return true;
      }

      // There were tasks in the queue. Wait a little bit more until no tasks are queued for the quiet period or
      // terminate if the quiet period is 0.
      // See https://github.com/netty/netty/issues/4241
      if (gracefulShutdownQuietPeriod == 0) {
        return true;
      }
      wakeup(true);
      return false;
    }

    final long nanoTime = ScheduledFutureTask.nanoTime();

    if (isShutdown() || nanoTime - gracefulShutdownStartTime > gracefulShutdownTimeout) {
      return true;
    }

    if (nanoTime - lastExecutionTime <= gracefulShutdownQuietPeriod) {
      // Check if any tasks were added to the queue every 100ms.
      // TODO: Change the behavior of takeTask() so that it returns on timeout.
      wakeup(true);
      try {
        Thread.sleep(100);
      } catch (InterruptedException e) {
        // Ignore
      }

      return false;
    }

    // No tasks were added for last quiet period - hopefully safe to shut down.
    // (Hopefully because we really cannot make a guarantee that there will be no execute() calls by a user.)
    return true;
  }

  @Override
  public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
    if (unit == null) {
      throw new NullPointerException("unit");
    }

    if (inEventLoop()) {
      throw new IllegalStateException("cannot await termination of the current thread");
    }

    if (threadLock.tryAcquire(timeout, unit)) {
      threadLock.release();
    }

    return isTerminated();
  }

  @Override
  public void execute(Runnable task) {
    if (task == null) {
      throw new NullPointerException("task");
    }

    boolean inEventLoop = inEventLoop();
    addTask(task);
    if (!inEventLoop) {
      startThread();
      if (isShutdown()) {
        boolean reject = false;
        try {
          if (removeTask(task)) {
            reject = true;
          }
        } catch (UnsupportedOperationException e) {
          // The task queue does not support removal so the best thing we can do is to just move on and
          // hope we will be able to pick-up the task before its completely terminated.
          // In worst case we will log on termination.
        }
        if (reject) {
          reject();
        }
      }
    }

    if (!addTaskWakesUp && wakesUpForTask(task)) {
      wakeup(inEventLoop);
    }
  }

  @Override
  public <T> T invokeAny(Collection<? extends Callable<T>> tasks)
      throws InterruptedException, ExecutionException {
    throwIfInEventLoop("invokeAny");
    return super.invokeAny(tasks);
  }

  @Override
  public <T> T invokeAny(Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
      throws InterruptedException, ExecutionException, TimeoutException {
    throwIfInEventLoop("invokeAny");
    return super.invokeAny(tasks, timeout, unit);
  }

  @Override
  public <T> List<java.util.concurrent.Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)
      throws InterruptedException {
    throwIfInEventLoop("invokeAll");
    return super.invokeAll(tasks);
  }

  @Override
  public <T> List<java.util.concurrent.Future<T>> invokeAll(
      Collection<? extends Callable<T>> tasks, long timeout, TimeUnit unit)
      throws InterruptedException {
    throwIfInEventLoop("invokeAll");
    return super.invokeAll(tasks, timeout, unit);
  }

  private void throwIfInEventLoop(String method) {
    if (inEventLoop()) {
      throw new RejectedExecutionException(
          "Calling " + method + " from within the EventLoop is not allowed");
    }
  }

  /**
   * Returns the {@link ThreadProperties} of the {@link Thread} that powers the {@link
   * SingleThreadEventExecutor}. If the {@link SingleThreadEventExecutor} is not started yet, this
   * operation will start it and block until it is fully started.
   */
  public final ThreadProperties threadProperties() {
    ThreadProperties threadProperties = this.threadProperties;
    if (threadProperties == null) {
      Thread thread = this.thread;
      if (thread == null) {
        assert !inEventLoop();
        submit(NOOP_TASK).syncUninterruptibly();
        thread = this.thread;
        assert thread != null;
      }

      threadProperties = new DefaultThreadProperties(thread);
      if (!PROPERTIES_UPDATER.compareAndSet(this, null, threadProperties)) {
        threadProperties = this.threadProperties;
      }
    }

    return threadProperties;
  }

  @SuppressWarnings("unused")
  protected boolean wakesUpForTask(Runnable task) {
    return true;
  }

  protected static void reject() {
    throw new RejectedExecutionException("event executor terminated");
  }

  /**
   * Offers the task to the associated {@link RejectedExecutionHandler}.
   *
   * @param task to reject.
   */
  protected final void reject(Runnable task) {
    rejectedExecutionHandler.rejected(task, this);
  }

  // ScheduledExecutorService implementation

  private static final long SCHEDULE_PURGE_INTERVAL = TimeUnit.SECONDS.toNanos(1);

  private void startThread() {
    if (state == ST_NOT_STARTED) {
      if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
        try {
          doStartThread();
        } catch (Throwable cause) {
          STATE_UPDATER.set(this, ST_NOT_STARTED);
          PlatformDependent.throwException(cause);
        }
      }
    }
  }

  private boolean ensureThreadStarted(int oldState) {
    if (oldState == ST_NOT_STARTED) {
      try {
        doStartThread();
      } catch (Throwable cause) {
        STATE_UPDATER.set(this, ST_TERMINATED);
        terminationFuture.tryFailure(cause);

        if (!(cause instanceof Exception)) {
          // Also rethrow as it may be an OOME for example
          PlatformDependent.throwException(cause);
        }
        return true;
      }
    }
    return false;
  }

  private void doStartThread() {
    assert thread == null;
    executor.execute(new Runnable() {
      @Override
      public void run() {
        thread = Thread.currentThread();
        if (interrupted) {
          thread.interrupt();
        }

        boolean success = false;
        updateLastExecutionTime();
        try {
          SingleThreadEventExecutor.this.run();
          success = true;
        } catch (Throwable t) {
          logger.warn("Unexpected exception from an event executor: ", t);
        } finally {
          for (; ; ) {
            int oldState = state;
            if (oldState >= ST_SHUTTING_DOWN || STATE_UPDATER.compareAndSet(
                SingleThreadEventExecutor.this, oldState, ST_SHUTTING_DOWN)) {
              break;
            }
          }

          // Check if confirmShutdown() was called at the end of the loop.
          if (success && gracefulShutdownStartTime == 0) {
            if (logger.isErrorEnabled()) {
              logger.error("Buggy " + EventExecutor.class.getSimpleName() + " implementation; " +
                  SingleThreadEventExecutor.class.getSimpleName() + ".confirmShutdown() must " +
                  "be called before run() implementation terminates.");
            }
          }

          try {
            // Run all remaining tasks and shutdown hooks.
            for (; ; ) {
              if (confirmShutdown()) {
                break;
              }
            }
          } finally {
            try {
              cleanup();
            } finally {
              // Lets remove all FastThreadLocals for the Thread as we are about to terminate and notify
              // the future. The user may block on the future and once it unblocks the JVM may terminate
              // and start unloading classes.
              // See https://github.com/netty/netty/issues/6596.
              FastThreadLocal.removeAll();

              STATE_UPDATER.set(SingleThreadEventExecutor.this, ST_TERMINATED);
              threadLock.release();
              if (!taskQueue.isEmpty()) {
                if (logger.isWarnEnabled()) {
                  logger.warn("An event executor terminated with " +
                      "non-empty task queue (" + taskQueue.size() + ')');
                }
              }
              terminationFuture.setSuccess(null);
            }
          }
        }
      }
    });
  }

  private static final class DefaultThreadProperties implements ThreadProperties {

    private final Thread t;

    DefaultThreadProperties(Thread t) {
      this.t = t;
    }

    @Override
    public State state() {
      return t.getState();
    }

    @Override
    public int priority() {
      return t.getPriority();
    }

    @Override
    public boolean isInterrupted() {
      return t.isInterrupted();
    }

    @Override
    public boolean isDaemon() {
      return t.isDaemon();
    }

    @Override
    public String name() {
      return t.getName();
    }

    @Override
    public long id() {
      return t.getId();
    }

    @Override
    public StackTraceElement[] stackTrace() {
      return t.getStackTrace();
    }

    @Override
    public boolean isAlive() {
      return t.isAlive();
    }
  }
}
